package com.navinfo.opentsp.platform.computing.serializer;

import com.google.common.base.Charsets;
import com.navinfo.opentsp.platform.computing.util.DateUtils;
import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
import com.navinfo.opentsp.platform.location.protocol.common.RealTimeDataPb;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.serialization.EventSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;

/**
 * D009 驾驶行为数据上传
 * HDFS_Sink的RealTimeData序列化器，拼接成字段间通过\001分割的字符串
 */
public class HDFSIncidentAlarmSerializer implements EventSerializer {
    private static final Logger logger = LoggerFactory.getLogger(HDFSIncidentAlarmSerializer.class);

    private final String APPEND_NEWLINE = "appendNewline";
    private final boolean appendNewline;
    private final String splitStr = "\001";
    private final OutputStream out;

    private HDFSIncidentAlarmSerializer(OutputStream out, Context ctx) {
        this.appendNewline = ctx.getBoolean(APPEND_NEWLINE, true);
        this.out = out;
    }

    public boolean supportsReopen() {
        return false;
    }

    public void afterCreate() {}

    public void afterReopen() {}

    public void beforeClose() {}

    public void flush() throws IOException {
        this.out.flush();
    }

    public void write(Event event) throws IOException {
        try {
            //拼写落盘到HDFS的字符串。字段间分隔符是\001，null通过\N表示
            LCIncidentAlarmData.IncidentAlarmData incidentAlarmData = LCIncidentAlarmData.IncidentAlarmData.parseFrom(event.getBody());
            StringBuilder locationStr = new StringBuilder(10000);
            locationStr
                    .append(incidentAlarmData.getTerminalId()).append(splitStr)
                    .append(incidentAlarmData.getSerialNumber()).append(splitStr)
                    .append(incidentAlarmData.getReceiveDate()).append(splitStr)
                    .append(incidentAlarmData.getGpsTime()).append(splitStr)
                    .append(incidentAlarmData.getLatitude()).append(splitStr)
                    .append(incidentAlarmData.getLongitude()).append(splitStr)
                    .append(incidentAlarmData.getOriginalLat()).append(splitStr)
                    .append(incidentAlarmData.getOriginalLng()).append(splitStr)
                    .append(incidentAlarmData.getHeight()).append(splitStr)
                    .append(incidentAlarmData.getDirection()).append(splitStr)
                    .append(incidentAlarmData.getDepotLogo()).append(splitStr)

                    //RapidAcceleration 急加速事件ID1信息
                    .append(incidentAlarmData.getRapidAcceleration().getDuration()).append(splitStr)
                    .append(incidentAlarmData.getRapidAcceleration().getRapidAccelerationFuelConsumption()).append(splitStr)
                    .append(incidentAlarmData.getRapidAcceleration().getRapidStatus()).append(splitStr)
                    //RapidDeceleration 急减速事件ID2信息
                    .append(incidentAlarmData.getRapidDeceleration().getDuration()).append(splitStr)
                    .append(incidentAlarmData.getRapidDeceleration().getRapidDecelerationStatus()).append(splitStr)
                    //OverSpeed 超速驾驶事件ID3信息
                    .append(incidentAlarmData.getOverSpeed().getFuelConsumption()).append(splitStr)
                    .append(incidentAlarmData.getOverSpeed().getOverSpeedStatus()).append(splitStr)
                    .append(incidentAlarmData.getOverSpeed().getDuration()).append(splitStr)
                    //SuperLongIdle 超长怠速事件ID4信息
                    .append(incidentAlarmData.getSuperLongIdle().getSuperLongIdleStatus()).append(splitStr)
                    .append(incidentAlarmData.getSuperLongIdle().getDuration()).append(splitStr)
                    .append(incidentAlarmData.getSuperLongIdle().getCumulativeFuelConsumption()).append(splitStr)
                    //IdleSpeedAirConditioner 怠速空调事件ID5信息
                    .append(incidentAlarmData.getIdleSpeedAirConditioner().getIdleSpeedAirConditionerStatus()).append(splitStr)
                    .append(incidentAlarmData.getIdleSpeedAirConditioner().getDuration()).append(splitStr)
                    .append(incidentAlarmData.getIdleSpeedAirConditioner().getCumulativeFuelConsumption()).append(splitStr)
                    //BigThrottle 大油门事件ID6信息
                    .append(incidentAlarmData.getBigThrottle().getBigThrottleStatus()).append(splitStr)
                    .append(incidentAlarmData.getBigThrottle().getDuration()).append(splitStr)
                    .append(incidentAlarmData.getBigThrottle().getCumulativeFuelConsumption()).append(splitStr)
                    //NeutralTaxi 空挡滑行事件ID7信息
                    .append(incidentAlarmData.getNeutralTaxi().getNeutralTaxiStatus()).append(splitStr)
                    .append(incidentAlarmData.getNeutralTaxi().getDuration()).append(splitStr)

                    //FatigueDriving 疲劳驾驶事件ID8信息
                    .append(incidentAlarmData.getFatigueDriving().getFatigueDrivingTime()).append(splitStr)
                    .append(incidentAlarmData.getFatigueDriving().getFatigueDrivingStatus()).append(splitStr)


                    //添加字段只能在分割线之上添加
                    //------------------------------------------分割线----------------------------------------
                    .append(DateUtils.format(incidentAlarmData.getGpsTime()* 1000, "yyyyMMdd")).append(splitStr);


            //将D009报文拼接后写入hdfs
            this.out.write(locationStr.toString().getBytes(Charsets.UTF_8));
            if (this.appendNewline) {
                this.out.write(10);
            }
        }catch (Exception e){
            logger.error("执行HDFS的IncidentAlarm拦截器失败：" + e.getMessage(), e);
        }
    }

    public static class Builder implements EventSerializer.Builder {
        public Builder() { }
        public EventSerializer build(Context context, OutputStream out) {
            return  new HDFSIncidentAlarmSerializer(out, context);
        }
    }
}
